package streamAPI.operators;

import com.google.common.collect.Lists;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import util.model.People;

import java.util.List;

/**
 * @author yue.cao
 * @since 09-28-2020
 */
public class DemoWithReduce {

	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		List<People> data = Lists.newArrayList(
				new People("cy", 20),
				new People("jack", 21),
				new People("mary", 22),
				new People("cy", 80)
		);

		DataStreamSource<People> source = env.fromCollection(data);

		source.keyBy(new KeySelector<People, String>() {
			@Override
			public String getKey(People value) throws Exception {
				return value.getName();
			}
		}).reduce(new ReduceFunction<People>() {
			@Override
			public People reduce(People value1, People value2) throws Exception {
				return new People(value1.getName(), value1.getAge() + value2.getAge());
			}
		}).print();

		env.execute();
	}

}
